package com.eshipenet.shipnet.edgechannelbuilder

import java.util.Properties

import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}
import play.api.libs.json.{JsValue, Json}

import scala.util.Try

object Main {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val jsValueStream = createEdgeBoxStream(env)
    val edgeChannelStream = jsValueStream
      .map(jsValue => DebeziumParser(jsValue))
      .filter(deb => deb.isSuccess)
      .map(_.get)
      .filter(_.isDefined)
      .map(_.get)
      .map(action => EdgeBoxAction(action))
      .filter(_.isSuccess)
      .map(_.get)
      .flatMap(_.genChannelAction)
    edgeChannelStream.print()

    val kafkaSink =
      new FlinkKafkaProducer[EdgeChannelAction]("demo.eshipe.net:9092",
                                                "EdgeChannel",
                                                new EdgeChannelActionKafkaSerializer())
    edgeChannelStream.addSink(kafkaSink)
    env.execute()
  }

  private def createEdgeBoxStream(env: StreamExecutionEnvironment): DataStream[JsValue] = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "demo.eshipe.net:9092")
    // only required for Kafka 0.8
    //    properties.setProperty("zookeeper.connect", "localhost:2181")
    properties.setProperty("group.id", "EdgeBox")
    val der =
      new EdgeBoxKafkaDeserializer()
    val stream = env
      .addSource(new FlinkKafkaConsumer[Tuple2[String, String]]("EdgeBox", der, properties))
      .map(tuple => (tuple.f0.toString, tuple.f1.toString))
      .map { tuple =>
        Try(Json.parse(tuple._2))
      }
      .filter(_.isSuccess)
      .map(_.get)
    //      .map(m => (m.id, m))
    stream
  }

}
